package io.plus.engine.pulse;

import io.plus.engine.client.ClientAsynchronousSocketChannelContext;
import io.plus.engine.client.ClientConfiguration;
import io.plus.engine.client.ReConnectConfiguration;
import io.plus.engine.pulse.task.CloseCode;
import io.plus.engine.server.ServerConfiguration;
import io.plus.engine.utils.lock.ReadLockHandler;
import io.plus.engine.utils.lock.ReadWriteLockMap;
import io.plus.engine.utils.lock.ReadWriteLockSet;
import lombok.extern.slf4j.Slf4j;

import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;


/**
 * 用户关心的API几乎全在这
 */
@Slf4j
public class Io {

    private Io() {
    }

    /**
     * 绑定业务id
     */
    public static void bindBusinessId(String businessId, AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.businessIds.bind(businessId, asynchronousSocketChannelContext);
    }

    /**
     * 绑定群组
     */
    public static void bindGroup(String groupId, AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.groups.bind(groupId, asynchronousSocketChannelContext);
    }

    /**
     * 将用户绑定到群组
     */
    public static void bindGroup(Configuration configuration, String userId, String groupId) {
        ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = Io.getByUserId(configuration, userId);
        if (setWithLock != null) {
            setWithLock.handle(new ReadLockHandler<Set<AsynchronousSocketChannelContext>>() {
                @Override
                public void handler(Set<AsynchronousSocketChannelContext> set) {
                    for (AsynchronousSocketChannelContext asynchronousSocketChannelContext : set) {
                        Io.bindGroup(groupId, asynchronousSocketChannelContext);
                    }
                }
            });
        }
    }

    /**
     * 绑定token
     */
    public static void bindToken(String token, AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.tokens.bind(token, asynchronousSocketChannelContext);
    }

    /**
     * 绑定用户
     */
    public static void bindUser(String userId, AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.users.bind(userId, asynchronousSocketChannelContext);
    }

    /**
     * 阻塞发送消息到指定ChannelContext
     */
    public static Boolean blockSend(AsynchronousSocketChannelContext asynchronousSocketChannelContext, PacketMetadata packetMetadata) {
        if (asynchronousSocketChannelContext == null) {
            return false;
        }
        CountDownLatch countDownLatch = new CountDownLatch(1);
        return send(asynchronousSocketChannelContext, packetMetadata, countDownLatch, PacketSendMode.single_packet_block_mode);
    }

    /**
     * 发送到指定的ip和port
     */
    public static Boolean blockSend(Configuration configuration, String ip, int port, PacketMetadata packetMetadata) {
        return send(configuration, ip, port, packetMetadata, true);
    }

    /**
     * 发消息到所有连接
     */
    public static Boolean blockSendToAll(Configuration configuration, PacketMetadata packetMetadata) {
        return sendToAll(configuration, packetMetadata, true);
    }

    /**
     * 阻塞发消息给指定业务ID
     */
    public static Boolean blockSendToBusinessId(Configuration configuration, String businessId, PacketMetadata packetMetadata) {
        return sendToBusinessId(configuration, businessId, packetMetadata, true);
    }

    /**
     * 发消息到组
     */
/*    public static Boolean blockSendToGroup(Configuration configuration, String group, Packet packet) {
        return blockSendToGroup(configuration, group, packet);
    }*/

    /**
     * 发消息到组
     */
    public static Boolean blockSendToGroup(Configuration configuration, String groupId, PacketMetadata packetMetadata) {
        return sendToGroup(configuration, groupId, packetMetadata, true);
    }

    /**
     * 发消息给指定ChannelContext packetId
     */
    public static Boolean blockSendToId(Configuration configuration, String id, PacketMetadata packetMetadata) {
        return sendToId(configuration, id, packetMetadata, true);
    }

    /**
     * 阻塞发送到指定ip对应的集合
     */
/*    public static Boolean blockSendToIp(Configuration configuration, String ip, Packet packet) {
        return blockSendToIp(configuration, ip, packet);
    }*/

    /**
     * 阻塞发送到指定ip对应的集合
     */
    public static Boolean blockSendToIp(Configuration configuration, String ip, PacketMetadata packetMetadata) {
        return sendToIp(configuration, ip, packetMetadata, true);
    }

    /**
     * 发消息到指定集合
     */
    public static Boolean blockSendToSet(Configuration configuration, ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock, PacketMetadata packetMetadata) {
        return sendToSet(configuration, setWithLock, packetMetadata, true);
    }

    /**
     * 阻塞发消息到指定token
     */
    public static Boolean blockSendToToken(Configuration configuration, String token, PacketMetadata packetMetadata) {
        return sendToToken(configuration, token, packetMetadata, true);
    }

    /**
     * 阻塞发消息给指定用户
     */
    public static Boolean blockSendToUser(Configuration configuration, String userId, PacketMetadata packetMetadata) {
        return sendToUser(configuration, userId, packetMetadata, true);
    }

    /**
     * 关闭连接
     */
    public static void close(AsynchronousSocketChannelContext asynchronousSocketChannelContext, String remark) {
        close(asynchronousSocketChannelContext, null, remark);
    }

    public static void close(AsynchronousSocketChannelContext asynchronousSocketChannelContext, String remark, CloseCode closeCode) {
        close(asynchronousSocketChannelContext, null, remark, closeCode);
    }

    /**
     * 关闭连接
     */
    public static void close(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Throwable throwable, String remark) {
        close(asynchronousSocketChannelContext, throwable, remark, false);
    }

    public static void close(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Throwable throwable, String remark, CloseCode closeCode) {
        close(asynchronousSocketChannelContext, throwable, remark, false, closeCode);
    }

    public static void close(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Throwable throwable, String remark, boolean isNeedRemove) {
        close(asynchronousSocketChannelContext, throwable, remark, isNeedRemove, true);
    }

    public static void close(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Throwable throwable, String remark, boolean isNeedRemove, CloseCode closeCode) {
        close(asynchronousSocketChannelContext, throwable, remark, isNeedRemove, true, closeCode);
    }

    public static void close(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Throwable throwable, String remark, boolean isNeedRemove, boolean needCloseLock) {
        close(asynchronousSocketChannelContext, throwable, remark, isNeedRemove, needCloseLock, null);
    }

    public static void close(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Throwable throwable, String remark, boolean isNeedRemove, boolean needCloseLock, CloseCode closeCode) {
        if (asynchronousSocketChannelContext == null) {
            return;
        }
        if (asynchronousSocketChannelContext.isWaitingClose) {
            log.debug("{} 正在等待被关闭", asynchronousSocketChannelContext);
            return;
        }

        //先立即取消各项任务，这样可防止有新的任务被提交进来
        asynchronousSocketChannelContext.decodeTask.setCanceled(true);
        asynchronousSocketChannelContext.handlerTask.setCanceled(true);
        asynchronousSocketChannelContext.sendTask.setCanceled(true);

        WriteLock writeLock = null;
        if (needCloseLock) {
            writeLock = asynchronousSocketChannelContext.closeLock.writeLock();

            boolean tryLock = writeLock.tryLock();
            if (!tryLock) {
                return;
            }
            asynchronousSocketChannelContext.isWaitingClose = true;
            writeLock.unlock();
        } else {
            asynchronousSocketChannelContext.isWaitingClose = true;
        }

        if (closeCode == null) {
            if (asynchronousSocketChannelContext.getCloseCode() == CloseCode.INIT_STATUS) {
                asynchronousSocketChannelContext.setCloseCode(CloseCode.NO_CODE);
            }
        } else {
            asynchronousSocketChannelContext.setCloseCode(closeCode);
        }

        if (asynchronousSocketChannelContext.asynchronousSocketChannel != null) {
            try {
                asynchronousSocketChannelContext.asynchronousSocketChannel.shutdownInput();
            } catch (Throwable e) {
                //log.error(e.toString(), e);
            }
            try {
                asynchronousSocketChannelContext.asynchronousSocketChannel.shutdownOutput();
            } catch (Throwable e) {
                //log.error(e.toString(), e);
            }
            try {
                asynchronousSocketChannelContext.asynchronousSocketChannel.close();
            } catch (Throwable e) {
                //log.error(e.toString(), e);
            }
        }

        asynchronousSocketChannelContext.closeTag.setRemark(remark);
        asynchronousSocketChannelContext.closeTag.setThrowable(throwable);
        if (!isNeedRemove) {
            if (asynchronousSocketChannelContext.isServer()) {
                isNeedRemove = true;
            } else {
                ClientAsynchronousSocketChannelContext clientChannelContext = (ClientAsynchronousSocketChannelContext) asynchronousSocketChannelContext;
                if (!ReConnectConfiguration.isNeedReConnect(clientChannelContext, false)) { //不需要重连
                    isNeedRemove = true;
                }
            }
        }
        asynchronousSocketChannelContext.closeTag.setNeedRemove(isNeedRemove);

        asynchronousSocketChannelContext.clientConfiguration.closeTask.add(asynchronousSocketChannelContext);
        asynchronousSocketChannelContext.clientConfiguration.closeTask.execute();
    }

    /**
     * 关闭连接
     */
    public static void close(Configuration configuration, String clientIp, Integer clientPort, Throwable throwable, String remark) {
        AsynchronousSocketChannelContext asynchronousSocketChannelContext = configuration.clientNodes.find(clientIp, clientPort);
        close(asynchronousSocketChannelContext, throwable, remark);
    }

    /**
     * 关闭某群所有连接
     */
    public static void closeGroup(Configuration configuration, String groupId, String remark) {
        closeGroup(configuration, groupId, remark, null);
    }

    /**
     * 关闭某群所有连接
     */
    public static void closeGroup(Configuration configuration, String groupId, String remark, CloseCode closeCode) {
        ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = Io.getChannelContextsByGroupId(configuration, groupId);
        setWithLock.handle(new ReadLockHandler<Set<AsynchronousSocketChannelContext>>() {
            @Override
            public void handler(Set<AsynchronousSocketChannelContext> set) {
                for (AsynchronousSocketChannelContext asynchronousSocketChannelContext : set) {
                    Io.close(asynchronousSocketChannelContext, remark, closeCode);
                }
            }
        });
    }

    /**
     * 获取所有连接，包括当前处于断开状态的
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getAll(Configuration configuration) {
        return configuration.connections;
    }

    /**
     * 获取所有连接，包括当前处于断开状态的
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getAllChannelContexts(Configuration configuration) {
        return getAll(configuration);
    }

    /**
     * 此API仅供 tio client使用
     * 获取所有处于正常连接状态的连接
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getConnecteds(ClientConfiguration clientConfiguration) {
        return clientConfiguration.connecteds;
    }

    /**
     * 此API仅供 tio client使用
     * 获取所有处于正常连接状态的连接
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getAllConnectedsChannelContexts(ClientConfiguration clientConfiguration) {
        return getConnecteds(clientConfiguration);
    }

    /**
     * 根据业务id找ChannelContext
     */
    public static AsynchronousSocketChannelContext getBusinessId(Configuration configuration, String businessId) {
        return configuration.businessIds.find(configuration, businessId);
    }

    /**
     * 根据业务id找ChannelContext
     */
    public static AsynchronousSocketChannelContext getChannelContextByBusinessId(Configuration configuration, String businessId) {
        return getBusinessId(configuration, businessId);
    }

    /**
     * 根据clientip和clientport获取ChannelContext
     */
    public static AsynchronousSocketChannelContext getByClientNode(Configuration configuration, String clientIp, Integer clientPort) {
        return configuration.clientNodes.find(clientIp, clientPort);
    }

    /**
     * 根据clientip和clientport获取ChannelContext
     */
    public static AsynchronousSocketChannelContext getChannelContextByClientNode(Configuration configuration, String clientIp, Integer clientPort) {
        return getByClientNode(configuration, clientIp, clientPort);
    }

    /**
     * 根据ChannelContext.id获取ChannelContext
     */
    public static AsynchronousSocketChannelContext getById(Configuration configuration, String id) {
        return configuration.ids.find(configuration, id);
    }

    /**
     * 根据ChannelContext.id获取ChannelContext
     */
    public static AsynchronousSocketChannelContext getChannelContextById(Configuration configuration, String id) {
        return getById(configuration, id);
    }

    /**
     * 获取一个组的所有客户端
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getByGroup(Configuration configuration, String groupId) {
        return configuration.groups.getChannelContextSet(configuration, groupId);
    }

    /**
     * 获取一个组的所有客户端
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getChannelContextsByGroupId(Configuration configuration, String groupId) {
        return getByGroup(configuration, groupId);
    }

    /**
     * 根据token获取SetWithLock<AsynchronousSocketChannelContext>
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getByToken(Configuration configuration, String token) {
        return configuration.tokens.find(configuration, token);
    }

    /**
     * 根据token获取SetWithLock<AsynchronousSocketChannelContext>
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getChannelContextsByToken(Configuration configuration, String token) {
        return getByToken(configuration, token);
    }

    /**
     * 根据userid获取SetWithLock<AsynchronousSocketChannelContext>
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getByUserId(Configuration configuration, String userId) {
        return configuration.users.find(configuration, userId);
    }

    /**
     * 根据userid获取SetWithLock<AsynchronousSocketChannelContext>
     */
    public static ReadWriteLockSet<AsynchronousSocketChannelContext> getChannelContextsByUserId(Configuration configuration, String userId) {
        return getByUserId(configuration, userId);
    }


    /**
     * 群组有多少个连接
     */
    public static int groupCount(Configuration configuration, String groupId) {
        ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = configuration.groups.getChannelContextSet(configuration, groupId);
        if (setWithLock == null) {
            return 0;
        }

        Set<AsynchronousSocketChannelContext> set = setWithLock.getObject();
        if (set == null) {
            return 0;
        }

        return set.size();
    }

    /**
     * 某通道是否在某群组中
     */
    public static boolean isInGroup(String groupId, AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        ReadWriteLockSet<String> setWithLock = asynchronousSocketChannelContext.getGroupIdSet();
        if (setWithLock == null) {
            return false;
        }

        Set<String> set = setWithLock.getObject();
        if (set == null) {
            return false;
        }

        return set.contains(groupId);
    }

    public static void remove(AsynchronousSocketChannelContext asynchronousSocketChannelContext, String remark) {
        remove(asynchronousSocketChannelContext, remark, null);
    }

    /**
     * 和close方法对应，只不过不再进行重连等维护性的操作
     */
    public static void remove(AsynchronousSocketChannelContext asynchronousSocketChannelContext, String remark, CloseCode closeCode) {
        remove(asynchronousSocketChannelContext, null, remark, closeCode);
    }

    /**
     * 和close方法对应，只不过不再进行重连等维护性的操作
     */
    public static void remove(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Throwable throwable, String remark) {
        remove(asynchronousSocketChannelContext, throwable, remark, null);
    }

    /**
     * 和close方法对应，只不过不再进行重连等维护性的操作
     */
    public static void remove(AsynchronousSocketChannelContext asynchronousSocketChannelContext, Throwable throwable, String remark, CloseCode closeCode) {
        close(asynchronousSocketChannelContext, throwable, remark, true, closeCode);
    }

    /**
     * 和close方法对应，只不过不再进行重连等维护性的操作
     */
    public static void remove(Configuration configuration, String clientIp, Integer clientPort, Throwable throwable, String remark) {
        remove(configuration, clientIp, clientPort, throwable, remark, null);
    }

    /**
     * 删除clientip和clientPort为指定值的连接
     */
    public static void remove(Configuration configuration, String clientIp, Integer clientPort, Throwable throwable, String remark, CloseCode closeCode) {
        AsynchronousSocketChannelContext asynchronousSocketChannelContext = configuration.clientNodes.find(clientIp, clientPort);
        remove(asynchronousSocketChannelContext, throwable, remark, closeCode);
    }

    /**
     * 删除clientip为指定值的所有连接
     */
    public static void remove(ServerConfiguration serverConfiguration, String ip, String remark) {
        remove(serverConfiguration, ip, remark, null);
    }

    /**
     * 删除clientip为指定值的所有连接
     */
    public static void remove(ServerConfiguration serverConfiguration, String ip, String remark, CloseCode closeCode) {
        ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = serverConfiguration.ips.getChannelContextSet(serverConfiguration, ip);
        if (setWithLock == null) {
            return;
        }

        setWithLock.handle(new ReadLockHandler<Set<AsynchronousSocketChannelContext>>() {
            @Override
            public void handler(Set<AsynchronousSocketChannelContext> set) {
                for (AsynchronousSocketChannelContext asynchronousSocketChannelContext : set) {
                    Io.remove(asynchronousSocketChannelContext, remark, closeCode);
                }
            }
        });
    }

    /**
     * 发送消息到指定ChannelContext
     */
    public static Boolean send(AsynchronousSocketChannelContext asynchronousSocketChannelContext, PacketMetadata packetMetadata) {
        return send(asynchronousSocketChannelContext, packetMetadata, null, null);
    }

    private static Boolean send(final AsynchronousSocketChannelContext asynchronousSocketChannelContext, PacketMetadata packetMetadata, CountDownLatch countDownLatch, PacketSendMode packetSendMode) {
        try {
            if (packetMetadata == null || asynchronousSocketChannelContext == null) {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                return false;
            }

            if (asynchronousSocketChannelContext.isVirtual) {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                return true;
            }

            if (asynchronousSocketChannelContext.isClosed || asynchronousSocketChannelContext.isRemoved) {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                if (asynchronousSocketChannelContext != null) {
                    log.info("can't send data, {}, isClosed:{}, isRemoved:{}", asynchronousSocketChannelContext, asynchronousSocketChannelContext.isClosed, asynchronousSocketChannelContext.isRemoved);
                }
                return false;
            }


            boolean isSingleBlock = countDownLatch != null && packetSendMode == PacketSendMode.single_packet_block_mode;

            boolean isAdded = false;
            if (countDownLatch != null) {
                PacketTag packetTag = new PacketTag();
                packetTag.setCountDownLatch(countDownLatch);
                packetMetadata.setPacketTag(packetTag);
            }


            if (asynchronousSocketChannelContext.clientConfiguration.useQueueSend) {
                isAdded = asynchronousSocketChannelContext.sendTask.add(packetMetadata);
            } else {
                isAdded = asynchronousSocketChannelContext.sendTask.sendPacket(packetMetadata);
            }

            if (!isAdded) {
                if (countDownLatch != null) {
                    countDownLatch.countDown();
                }
                return false;
            }
            if (asynchronousSocketChannelContext.clientConfiguration.useQueueSend) {
                asynchronousSocketChannelContext.sendTask.execute();
            }

            if (isSingleBlock) {
                long timeout = 10;
                try {
                    Boolean awaitFlag = countDownLatch.await(timeout, TimeUnit.SECONDS);
                    if (!awaitFlag) {
                        log.error("{}, 阻塞发送超时, timeout:{}s, packetMetadata:{}", asynchronousSocketChannelContext, timeout, packetMetadata.toString());
                    }
                } catch (InterruptedException e) {
                    log.error(e.toString(), e);
                }

                Boolean isSentSuccess = packetMetadata.getPacketTag().getIsSentSuccess();
                return isSentSuccess;
            } else {
                return true;
            }
        } catch (Throwable e) {
            log.error(asynchronousSocketChannelContext + ", " + e.toString(), e);
            return false;
        }

    }

    /**
     * 发送到指定的ip和port
     */
    public static Boolean send(Configuration configuration, String ip, int port, PacketMetadata packetMetadata) {
        return send(configuration, ip, port, packetMetadata, false);
    }

    /**
     * 发送到指定的ip和port
     */
    private static Boolean send(Configuration configuration, String ip, int port, PacketMetadata packetMetadata, boolean isBlock) {
        AsynchronousSocketChannelContext asynchronousSocketChannelContext = configuration.clientNodes.find(ip, port);
        if (asynchronousSocketChannelContext != null) {
            if (isBlock) {
                return blockSend(asynchronousSocketChannelContext, packetMetadata);
            } else {
                return send(asynchronousSocketChannelContext, packetMetadata);
            }
        } else {
            log.info("{}, can find clientChannelContext by {}:{}", configuration.getName(), ip, port);
            return false;
        }
    }


    /**
     * 发消息到所有连接
     */
    public static void sendToAll(Configuration configuration, PacketMetadata packetMetadata) {
        sendToAll(configuration, packetMetadata, false);
    }

    private static Boolean sendToAll(Configuration configuration, PacketMetadata packetMetadata, boolean isBlock) {
        try {
            ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = configuration.connections;
            if (setWithLock == null) {
                log.debug("{}, 没有任何连接", configuration.getName());
                return false;
            }
            Boolean ret = sendToSet(configuration, setWithLock, packetMetadata, isBlock);
            return ret;
        } finally {
        }
    }

    /**
     * 发消息给指定业务ID
     */
    public static Boolean sendToBusinessId(Configuration configuration, String businessId, PacketMetadata packetMetadata) {
        return sendToBusinessId(configuration, businessId, packetMetadata, false);
    }

    /**
     * 发消息给指定业务ID
     */
    private static Boolean sendToBusinessId(Configuration configuration, String businessId, PacketMetadata packetMetadata, boolean isBlock) {
        AsynchronousSocketChannelContext asynchronousSocketChannelContext = Io.getChannelContextByBusinessId(configuration, businessId);
        if (asynchronousSocketChannelContext == null) {

            return false;
        }
        if (isBlock) {
            return blockSend(asynchronousSocketChannelContext, packetMetadata);
        } else {
            return send(asynchronousSocketChannelContext, packetMetadata);
        }
    }


    /**
     * 发消息到组
     */
    public static void sendToGroup(Configuration configuration, String groupId, PacketMetadata packetMetadata) {
        sendToGroup(configuration, groupId, packetMetadata, false);
    }

    /**
     * 发消息到组
     */
    private static Boolean sendToGroup(Configuration configuration, String groupId, PacketMetadata packetMetadata, boolean isBlock) {
        try {
            ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = configuration.groups.getChannelContextSet(configuration, groupId);
            if (setWithLock == null) {
                log.debug("{}, 组[{}]不存在", configuration.getName(), groupId);
                return false;
            }
            Boolean ret = sendToSet(configuration, setWithLock, packetMetadata, isBlock);
            return ret;
        } finally {

        }
    }

    /**
     * 发消息给指定ChannelContext packetId
     */
    public static Boolean sendToId(Configuration configuration, String id, PacketMetadata packetMetadata) {
        return sendToId(configuration, id, packetMetadata, false);
    }

    /**
     * 发消息给指定ChannelContext packetId
     */
    private static Boolean sendToId(Configuration configuration, String channelContextId, PacketMetadata packetMetadata, boolean isBlock) {
        AsynchronousSocketChannelContext asynchronousSocketChannelContext = Io.getChannelContextById(configuration, channelContextId);
        if (asynchronousSocketChannelContext == null) {

            return false;
        }
        if (isBlock) {
            return blockSend(asynchronousSocketChannelContext, packetMetadata);
        } else {
            return send(asynchronousSocketChannelContext, packetMetadata);
        }
    }


    /**
     * 发送到指定ip对应的集合
     */
    public static void sendToIp(Configuration configuration, String ip, PacketMetadata packetMetadata) {
        sendToIp(configuration, ip, packetMetadata, false);
    }

    /**
     * 发送到指定ip对应的集合
     */
    private static Boolean sendToIp(Configuration configuration, String ip, PacketMetadata packetMetadata, boolean isBlock) {
        try {
            ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = configuration.ips.getChannelContextSet(configuration, ip);
            if (setWithLock == null) {
                log.info("{}, 没有ip为[{}]的对端", configuration.getName(), ip);
                return false;
            }
            Boolean ret = sendToSet(configuration, setWithLock, packetMetadata, isBlock);
            return ret;
        } finally {

        }
    }

    /**
     * 发消息到指定集合
     */
    public static void sendToSet(Configuration configuration, ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock, PacketMetadata packetMetadata) {
        sendToSet(configuration, setWithLock, packetMetadata, false);
    }

    /**
     * 发消息到指定集合
     */
    private static Boolean sendToSet(Configuration configuration, ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock, PacketMetadata packetMetadata, boolean isBlock) {
        boolean releasedLock = false;
        Lock lock = setWithLock.getReadLock();
        lock.lock();
        try {
            Set<AsynchronousSocketChannelContext> set = setWithLock.getObject();
            if (set.size() == 0) {
                log.debug("{}, 集合为空", configuration.getName());
                return false;
            }

            CountDownLatch countDownLatch = null;
            if (isBlock) {
                countDownLatch = new CountDownLatch(set.size());
            }
            int sendCount = 0;
            for (AsynchronousSocketChannelContext asynchronousSocketChannelContext : set) {
                sendCount++;
                if (isBlock) {
                    send(asynchronousSocketChannelContext, packetMetadata, countDownLatch, PacketSendMode.one_group_packet_block_mode);
                } else {
                    send(asynchronousSocketChannelContext, packetMetadata, null, null);
                }
            }
            lock.unlock();
            releasedLock = true;

            if (sendCount == 0) {
                return false;
            }

            if (isBlock) {
                try {
                    long timeout = sendCount / 5;
                    timeout = Math.max(timeout, 10);//timeout < 10 ? 10 : timeout;
                    boolean awaitFlag = countDownLatch.await(timeout, TimeUnit.SECONDS);
                    if (!awaitFlag) {
                        log.error("{}, 同步群发超时, size:{}, timeout:{}, packetMetadata:{}", configuration.getName(), setWithLock.getObject().size(), timeout, packetMetadata.toString());
                        return false;
                    } else {
                        return true;
                    }
                } catch (InterruptedException e) {
                    log.error(e.toString(), e);
                    return false;
                } finally {

                }
            } else {
                return true;
            }
        } catch (Throwable e) {
            log.error(e.toString(), e);
            return false;
        } finally {
            if (!releasedLock) {
                lock.unlock();
            }
        }
    }

    /**
     * 发消息到指定token
     */
    public static Boolean sendToToken(Configuration configuration, String token, PacketMetadata packetMetadata) {
        return sendToToken(configuration, token, packetMetadata, false);
    }

    /**
     * 发消息给指定token
     */
    private static Boolean sendToToken(Configuration configuration, String token, PacketMetadata packetMetadata, boolean isBlock) {
        ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = configuration.tokens.find(configuration, token);
        try {
            if (setWithLock == null) {
                return false;
            }

            ReadLock readLock = setWithLock.getReadLock();
            readLock.lock();
            try {
                Set<AsynchronousSocketChannelContext> set = setWithLock.getObject();
                boolean ret = false;
                for (AsynchronousSocketChannelContext asynchronousSocketChannelContext : set) {
                    boolean singleRet = false;
                    // 不要用 a = a || b()，容易漏执行后面的函数
                    if (isBlock) {
                        singleRet = blockSend(asynchronousSocketChannelContext, packetMetadata);
                    } else {
                        singleRet = send(asynchronousSocketChannelContext, packetMetadata);
                    }
                    if (singleRet) {
                        ret = true;
                    }
                }
                return ret;
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            } finally {
                readLock.unlock();
            }
            return false;
        } finally {

        }
    }

    /**
     * 发消息给指定用户
     */
    public static Boolean sendToUser(Configuration configuration, String userId, PacketMetadata packetMetadata) {
        return sendToUser(configuration, userId, packetMetadata, false);
    }

    /**
     * 发消息给指定用户
     */
    private static Boolean sendToUser(Configuration configuration, String userId, PacketMetadata packetMetadata, boolean isBlock) {
        ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = configuration.users.find(configuration, userId);
        try {
            if (setWithLock == null) {
                return false;
            }

            ReadLock readLock = setWithLock.getReadLock();
            readLock.lock();
            try {
                Set<AsynchronousSocketChannelContext> set = setWithLock.getObject();
                boolean ret = false;
                for (AsynchronousSocketChannelContext asynchronousSocketChannelContext : set) {
                    boolean singleRet = false;
                    // 不要用 a = a || b()，容易漏执行后面的函数
                    if (isBlock) {
                        singleRet = blockSend(asynchronousSocketChannelContext, packetMetadata);
                    } else {
                        singleRet = send(asynchronousSocketChannelContext, packetMetadata);
                    }
                    if (singleRet) {
                        ret = true;
                    }
                }
                return ret;
            } catch (Throwable e) {
                log.error(e.getMessage(), e);
            } finally {
                readLock.unlock();
            }
            return false;
        } finally {

        }
    }

    /**
     * 发送并等待响应.<br>
     * 注意：<br>
     * 1、发送消息时，设置一下synSeq，形如：xxPacket.setSynchronismSerialNumber(789798786)。synSeq不为空且大于0（null、等于小于0都不行）<br>
     * 2、对端收到此消息后，需要回一条synSeq一样的消息，所以业务需要在decode()方法中根据bytebuffer反解析出packet的synSeq值，并赋给XxPacket对象<br>
     * 3、对于同步发送，框架层面并不会帮应用去调用handler.handler(packetMetadata, clientChannelContext)方法，应用需要自己去处理响应的消息包，
     * 参考写法：clientConfiguration.getHandler().handler(packetMetadata, clientChannelContext);<br>
     *
     * @param asynchronousSocketChannelContext
     * @param packetMetadata                   业务层必须设置好synSeq字段的值，而且要保证唯一（不能重复）。可以在tioConfig范围内用AtomicInteger
     */
    @SuppressWarnings("finally")
    public static PacketMetadata synSend(AsynchronousSocketChannelContext asynchronousSocketChannelContext, PacketMetadata packetMetadata, long timeout) {
        Integer synchronismSerialNumber = packetMetadata.getSynchronismSerialNumber();
        if (synchronismSerialNumber == null || synchronismSerialNumber <= 0) {
            throw new RuntimeException("synSeq必须大于0");
        }

        ReadWriteLockMap<Integer, PacketMetadata> synchronismSerialNumberToPacketMap = asynchronousSocketChannelContext.clientConfiguration.getSynchronismSerialNumberToPacketMap();
        try {
            synchronismSerialNumberToPacketMap.put(synchronismSerialNumber, packetMetadata);

            synchronized (packetMetadata) {
                send(asynchronousSocketChannelContext, packetMetadata);
                try {
                    packetMetadata.wait(timeout);
                } catch (InterruptedException e) {
                    log.error(e.toString(), e);
                }
            }
        } catch (Throwable e) {
            log.error(e.toString(), e);
        } finally {
            PacketMetadata respPacketMetadata = synchronismSerialNumberToPacketMap.remove(synchronismSerialNumber);
            if (respPacketMetadata == null) {
                log.error("respPacketMetadata == null,{}", asynchronousSocketChannelContext);
                return null;
            }
            if (respPacketMetadata == packetMetadata) {
                log.error("{}, 同步发送超时, {}", asynchronousSocketChannelContext.clientConfiguration.getName(), asynchronousSocketChannelContext);
                return null;
            }
            return respPacketMetadata;
        }
    }

    /**
     * 解绑业务id
     */
    public static void unbindBsId(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.businessIds.unbind(asynchronousSocketChannelContext);
    }

    /**
     * 与所有组解除解绑关系
     */
    public static void unbindGroup(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.groups.unbind(asynchronousSocketChannelContext);
    }

    /**
     * 与指定组解除绑定关系
     */
    public static void unbindGroup(String groupId, AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.groups.unbind(groupId, asynchronousSocketChannelContext);
    }

    /**
     * 将某用户从组中解除绑定
     */
    public static void unbindGroup(Configuration configuration, String userId, String groupId) {
        ReadWriteLockSet<AsynchronousSocketChannelContext> setWithLock = Io.getByUserId(configuration, userId);
        if (setWithLock != null) {
            setWithLock.handle(new ReadLockHandler<Set<AsynchronousSocketChannelContext>>() {
                @Override
                public void handler(Set<AsynchronousSocketChannelContext> set) {
                    for (AsynchronousSocketChannelContext asynchronousSocketChannelContext : set) {
                        Io.unbindGroup(groupId, asynchronousSocketChannelContext);
                    }
                }
            });
        }
    }

    /**
     * 解除channelContext绑定的token
     */
    public static void unbindToken(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.tokens.unbind(asynchronousSocketChannelContext);
    }

    /**
     * 解除token
     */
    public static void unbindToken(Configuration configuration, String token) {
        configuration.tokens.unbind(configuration, token);
    }

    /**
     * 解除channelContext绑定的userid
     */
    public static void unbindUser(AsynchronousSocketChannelContext asynchronousSocketChannelContext) {
        asynchronousSocketChannelContext.clientConfiguration.users.unbind(asynchronousSocketChannelContext);
    }

    /**
     * 解除userid的绑定。一般用于多地登录，踢掉前面登录的场景
     */
    public static void unbindUser(Configuration configuration, String userId) {
        configuration.users.unbind(configuration, userId);
    }

}
